package com.bbx.flink.demo.windowfuncation;

import akka.dispatch.Foreach;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class ProcessWindowFunctionDemo {

    public static void main(String [] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.socketTextStream("192.168.10.131",10003)
                .map((MapFunction<String,Integer>) value->Integer.parseInt(value))
                .keyBy(i->i%2)
                .timeWindow(Time.minutes(1))
                .process(new ProcessWindowFunction<Integer, String, Integer, TimeWindow>() {
                    private static final long serialVersionUID = -4434788063241088499L;

                    /**
                     * @param integer     key
                     * @param context     content
                     * @param elements    input elements
                     * @param out         output
                     * @throws Exception
                     */
                    @Override
                    public void process(Integer integer, Context context, Iterable<Integer> elements, Collector<String> out) throws Exception {
                        int sum = 0;

                        for(Integer els : elements){
                            sum+=els;
                        }
                        out.collect("窗口key为："+integer+"，运算后的结果："+sum+",context为："+context);
                    }
                }).print();
        env.execute();
    }
}
